package net.bwie.realtime.vehicle2.function;

import net.bwie.realtime.vehicle2.bean.VehicleData;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Set;

/**
 * @Author: FuHe
 * @Date: 2025/6/8
 */
public class VehicleVinCountFunction extends ProcessAllWindowFunction<VehicleData, String, TimeWindow> {
    @Override
    public void process(ProcessAllWindowFunction<VehicleData, String, TimeWindow>.Context context, Iterable<VehicleData> elements, Collector<String> out) throws Exception {
        Set<String> hashSet = new HashSet<>();
        for (VehicleData element : elements) {
            if (element.getVin() != null && !hashSet.contains(element.getVin())) {
                hashSet.add(element.getVin());
            }
        }
        long start = context.window().getStart();
        long end = context.window().getEnd();
        int count = hashSet.size();
        out.collect(start + "," + end + "," + count);
    }
}
